/*
 * @Author: liuzelin
 * @Date: 2024-04-20 10:39:20
 * @LastEditors: liuzelin
 * @LastEditTime: 2024-05-20 22:21:59
 * @Description: file content
 */
#pragma once

#include <map>
#include <unordered_set>
#include <memory>
#include <mutex>
#include <queue>
#include <unordered_map>

#include "panda_event_impl.hpp"
#include "panda_thread.hpp"
#include "panda_timer.hpp"
#include "panda_uv_handler.hpp"
#include "panda_uv_helper.hpp"
#include "panda_queue.hpp"
#include "uv.h"

namespace PandaImpl {
class ThreadImpl;
class UVHelper;
class UVTimer : public Panda::Timer, public UVHandler {
  friend class UVHelper;

 public:
  explicit UVTimer(ThreadImpl *thread);
  void Start(uint64_t timeOutMs, OnTimer &&onTimer) override;
  void Stop() override;
  void OnClose() override {}
  ~UVTimer() = default;

 private:
  ThreadImpl *thread_;
  uv_timer_t timer_;
  OnTimer onTimer_;
};

struct TaskImpl {
  explicit TaskImpl( Panda::PandaTask&& task, Panda::TaskLocation location );
  TaskImpl() = default;
  Panda::PandaTask task;
  Panda::TaskLocation location;
  uint64_t create_us;
};

struct DelayTaskImpl : public TaskImpl {
  DelayTaskImpl(Panda::PandaTask &&task, Panda::TaskLocation location,
    uint64_t delay );
  DelayTaskImpl( DelayTaskImpl& other );
  DelayTaskImpl( DelayTaskImpl&& other );
  DelayTaskImpl() = default;
  uint64_t delay;
};

class ThreadImpl : public Panda::Thread {
  friend class UVHelper;

 public:
  ThreadImpl(bool main) : isMain_(main) {
    if (isMain_) this->PreRun();
  }
  ~ThreadImpl();
  void SetThreadName( const char* name ) override { this->name_ = name; }
  const char *Name() const override { return name_.c_str(); }
  void Start() override {
    if (isMain_) {
      this->Run();
    } else {
      if (isRunning_) return;
      thread_.reset(new std::thread([this] { this->Run(); }));
      while (this->isRunning_ == false) {
      }
    }
  }
  bool IsCurrent() const override;
  void Stop() override;

  Panda::UDPSocket *CreateUDPSocket() override;
  Panda::TCPSocket *CreateTCPSocket() override;
  Panda::Timer *CreateTimer() override;
  Panda::Resolver *CreateResolver() override;

  void PostTask(Panda::TaskLocation location, Panda::PandaTask func);
  void ClearDelayTask(void *key);
  void PostDelayTask(Panda::TaskLocation location, Panda::PandaTask func,
                     uint64_t delay, void *key = nullptr);

  uv_loop_t *GetLoop() { return &uvLoop_; }

  void DisableCalculateTaskCost();

 private:
  void PreRun();
  void Run();
  void Loop();
  void PostRun();
  std::string name_;
  bool isMain_ = false;
  volatile bool isRunning_ = false;
  uv_loop_t uvLoop_;
  std::unique_ptr<std::thread> thread_;
  std::unique_ptr<UVTimer> timer_;
  std::unique_ptr<UVAsync> event_;
  std::mutex tasksMutex_;
  std::queue<TaskImpl> tasks_;
  std::mutex delayTasksMutex_;
  typedef std::multimap<uint64_t, DelayTaskImpl> DelayTasks;
  typedef std::multimap<uint64_t, DelayTaskImpl>::iterator DelayTask;
  std::unordered_map<void *, DelayTasks> delayTasks_;
};

struct TaskLocationHash {
  std::size_t operator () ( Panda::TaskLocation const & location ) const {
    std::size_t h1 = std::hash<std::string>{}( location.File() );
    std::size_t h2 = std::hash<std::string>{}( location.Function() );
    return h1 ^ h2;
  }
};

struct TaskLocationEqual {
  bool operator () ( const Panda::TaskLocation& a, const Panda::TaskLocation& b ) const {
    return a == b;
  }
};

class ThreadManagerImpl : public Panda::ThreadManager {
 public:
  ThreadManagerImpl() {}

  Panda::Thread *CreateThread() override { return new ThreadImpl(false); }
  Panda::Thread *Current() override;
  void EnableCalculateTaskCost(bool enable) override;
  void ExportTaskCostStat(const char *dir) override;
  void TaskAfterExec(Panda::Thread *thread, Panda::TaskLocation &location,
                     int64_t use_us);

 private:
  bool enable_calculate_task_cost = false;                
  struct TaskCost{
    Panda::TaskLocation task_location;
    Panda::Thread *thread;
    int64_t use_us;
    TaskCost& operator = (TaskCost& other) {
      this->thread = other.thread;
      this->use_us = other.use_us;
      this->task_location = other.task_location;
      return *this;
    }
  };
  Panda::LockFreeQueue<TaskCost> task_cost_que;
  std::unique_ptr<Panda::Thread> calculate_task_cost_thread;
  struct TaskCostStat {
    uint64_t total_use_us = 0;
    uint64_t total_call_num = 0; 
    struct ThreadTaskCostStat {
      uint64_t total_use_us = 0;
      uint64_t total_call_num = 0;
    };
    std::map<Panda::Thread *, ThreadTaskCostStat> thread_cost_stat;
  };
  std::unordered_map<Panda::TaskLocation, TaskCostStat, TaskLocationHash, TaskLocationEqual> task_cost_map;
};

class ThreadPoolManagerImpl : public Panda::ThreadPoolManager {
public:
  ThreadPoolManagerImpl() = default;
  void Start( int thread_num = 0 ) override;
  void PostTask( Panda::TaskLocation location, Panda::PandaTask func ) override;

private:
  struct ThreadHandler {
    explicit ThreadHandler( ThreadPoolManagerImpl* m, int index ) : manager( m ), cpu_index( index ) {
      thread.reset( new std::thread( [ this ]() {this->ThreadFunc();} ) );
    }
    ~ThreadHandler() {
      running = false;
      manager->dispatch_event.SetAll();
      thread->join();
    }
    void ThreadFunc();
    ThreadPoolManagerImpl* manager = nullptr;
    volatile bool running = false;
    int cpu_index = -1;
    std::unique_ptr<std::thread> thread;
  };
  PandaImpl::TaskImpl PollTask( ThreadHandler* handler );
  void AfterExecTask( Panda::TaskLocation location, uint64_t task_start, uint64_t task_create );
  Panda::LockFreeQueue<TaskImpl> tasks_;
  std::vector<std::unique_ptr<ThreadHandler>> threads;
  std::unique_ptr<ThreadImpl> dispatch_thread;
  STDEvent dispatch_event;
  struct TaskExecInfo {
    uint64_t total_use_us = 0;
    uint64_t total_call_num = 0; 
    uint64_t total_exec_delay = 0;
  };
  std::unordered_map<Panda::TaskLocation, TaskExecInfo, TaskLocationHash, TaskLocationEqual> task_exec_info;
};
}  // namespace PandaImpl